package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.JsonResult;
import cn.tedu.charging.common.pojo.param.ChargingProcessParam;
import cn.tedu.charging.common.pojo.vo.ChargingProcessVO;
import cn.tedu.charging.common.utils.JsonUtils;
import cn.tedu.charging.order.common.ChargingConstants;
import cn.tedu.charging.order.common.MqttConstants;
import cn.tedu.charging.order.common.OrderStatusConstants;
import cn.tedu.charging.order.pojo.dto.ChargingProcessDto;
import cn.tedu.charging.order.pojo.dto.ChargingResultDto;
import cn.tedu.charging.order.pojo.po.ChargingOrderFailPO;
import cn.tedu.charging.order.pojo.po.ChargingOrderSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    //@Autowired
    //private MqttClient mqttClient;

    private MqttClient mqttClient;

    private MqttContext mqttContext;

    /**
     * 没有在容器中的 MqttConsumer 想使用 容器中 MqttClient
     * 可以在创建 MqttConsumer 的时候，把 MqttClient 通过构造方法 传入进来
     * @param mqttClient
     */
    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    /**
     * 通过 setter 方法 设置 MqttClient
     * @param mqttClient
     */
    public void setMqttClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }



    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.debug("connectComplete 连接成功:reconnect:{},serverURI:{}",
                reconnect, serverURI);
        //在和EMQX连接成功后
        //订阅Topic 设备返回的开始充电结果Topic
        String topicChargingResult = MqttConstants.TOPIC_CHARGING_RESULT;
        //订阅topic  设备定时同步的充电进度的Topic
        String topicChargingProcess = MqttConstants.TOPIC_CHARGING_PROCESS;
        try {
            log.debug("订单服务订阅设备返回的 <<开始充电结果>> Topic:{}", topicChargingResult);
            mqttClient.subscribe(topicChargingResult);
            log.debug("订单服务订阅设备返回的 <<开始充电结果>> Topic:{},成功", topicChargingResult);

            log.debug("订单服务订阅设备返回的 <<同步充电进度>> Topic:{}", topicChargingProcess);
            mqttClient.subscribe(topicChargingProcess);
            log.debug("订单服务订阅设备返回的 <<同步充电进度>> Topic:{},成功", topicChargingProcess);

        } catch (MqttException e) {
            //打印异常信息 可能进行重试，接入告警系统 告警
            throw new RuntimeException(e);
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.debug("connectionLost 连接丢失:{}", cause);
        //在连接丢失的时候  重连
        //mqttClient.connect();
    }

    /**
     * 消息到达
     * 设备  给  订单服务 发送的消息 到达了
     * @param topic name of the topic on the message was published to
     * @param message the actual message.
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("messageArrived 收到设备发送的消息:topic:{},message:{}", topic, message);
        if (MqttConstants.TOPIC_CHARGING_PROCESS.equals(topic)) {
            log.debug("处理设备同步的充电进度消息");
            handleChargingProcess(message);
        }else if (MqttConstants.TOPIC_CHARGING_RESULT.equals(topic)) {
            log.debug("处理设备同步的开始充电结果消息");
            handleChargingResult(message);
        }
    }

    /**
     * 处理开始充电进度消息
     * @param message
     */
    private void handleChargingProcess(MqttMessage message) {
        ChargingProcessDto chargingProcessDto = null;
        try {
        chargingProcessDto = JsonUtils.fromJson(message.toString(),
                ChargingProcessDto.class);
        }catch (Exception e) {
             //如果多次失败，应该告警
             log.error("解析设备返回的消息失败:{}", e);
        }
        log.debug("收到设备同步的充电进度数据:{}", chargingProcessDto);
        if (chargingProcessDto != null) {
            //1 计算价格
            ChargingProcessVO cost = getCost(chargingProcessDto);
            //2 保存充电进度 ES
            saveChargingProcess(chargingProcessDto,cost);

            //3 判断用户余额
            /**
             * 设备同步的充电数据 会包含 是否充满
             * 1. 判断是否充满
             * 2. 如果没充满 继续
             * 3. 如果充满
             * 4. 修改订单状态 为 完成    订单服务  分布式事务
             * 5. 扣除用户余额                用户服务  分布式事务
             * 6. 修改订单状态 和 扣除用户余额 要么全部成功 要么全部失败 分布式事务
             * 7. 修改枪状态 为 空闲
             * 8. 可以扩展  给用户增加积分 或者 发送优惠券  营销
             * 9. 通知用户 用户账单 充了多长时间，花了多少钱 充了多少度，余额还有多少
             */

            //4 推送数据给用户
            sendChargingProcess2User(cost);

            //
            //5 判断设备温度
            /**
             * 设备同步的充电数据 会包含 设备的温度
             * 1. 分级 正常 20-30   告警 30-40 可以充，让用户选择是否继续  异常 大于 40 强制停止充电
             * 2. 判断温度是否正常
             * 3. 如果正常 继续充电
             * 4. 如果异常 通知用户 告警   让用户选择是否继续  异常 停止充电 设置状态为故障
             */

            //6 判断是否充满
            /**
             * 设备同步的充电数据 会包含 是否充满
             * 1. 判断是否充满
             * 2. 如果没充满 继续
             * 3. 如果充满
             * 4. 修改订单状态 为 完成    订单服务  分布式事务
             * 5. 扣除用户余额                用户服务  分布式事务
             * 6. 修改订单状态 和 扣除用户余额 要么全部成功 要么全部失败 分布式事务
             * 7. 修改枪状态 为 空闲
             * 8. 可以扩展  给用户增加积分 或者 发送优惠券  营销
             * 9. 通知用户 用户账单 充了多长时间，花了多少钱 充了多少度，余额还有多少
             */


        }
    }

    /**
     * 发送充电进度给用户
     * @param cost
     */
    private void sendChargingProcess2User(ChargingProcessVO cost) {
        String json = JsonUtils.toJson(cost);
        Integer userId = cost.getUserId();
        log.debug("同步充电数据,用户:{},消息:{}",userId,cost);
        mqttContext.getWebSocketServer().sendMessage(userId,json);
        log.debug("同步充电数据成功,用户:{},消息:{}",userId,cost);
    }

    /**
     * 保存充电进度到ES
     * 保存 充电进度的完整数据
     * 每次设备同步的数据 chargingProcessDto
     * 每次计算的价格 cost
     * 同步的时间
     * @param chargingProcessDto
     * @param cost
     */
    private void saveChargingProcess(ChargingProcessDto chargingProcessDto, ChargingProcessVO cost) {
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        BeanUtils.copyProperties(chargingProcessDto,chargingProcessPO);
        BeanUtils.copyProperties(cost,chargingProcessPO);
        //设置id 充电数据的时间
        chargingProcessPO.setId(System.currentTimeMillis());
        log.debug("保存充电进度数据到ES:{}",chargingProcessPO);
        mqttContext.getChargingProcessESRepository().save(chargingProcessPO);
        log.debug("保存充电进度数据到ES:{},成功",chargingProcessPO);
    }


    /**
     * 通过 Feign客户端 CostClient 调用计价服务计算价格
     * @param chargingProcessDto
     * @return
     */
    private ChargingProcessVO getCost(ChargingProcessDto chargingProcessDto){
        ChargingProcessParam chargingProcessParam = new ChargingProcessParam();
        //通过BeanUtils.copyProperties来复制属性值
        BeanUtils.copyProperties(chargingProcessDto,chargingProcessParam);
        log.debug("调用计价服务计算价格入参:{}",chargingProcessParam);
        JsonResult<ChargingProcessVO> result =
                mqttContext.getCostClient().calculateCost(chargingProcessParam);
        log.debug("调用计价服务计算价格入参:{},结果:{}",chargingProcessDto,result);
        if (result != null) {
            return result.getData();
        }
        return null;
    }


    /**
     * 处理开始充电结果消息
     * @param message
     */
    private void handleChargingResult(MqttMessage message) {
        //  new String(payload); 字节数组转换为字符串
        ChargingResultDto chargingResultDto = null;
        //try {
        chargingResultDto = JsonUtils.fromJson(message.toString(),
                ChargingResultDto.class);
        // }catch (Exception e) {
        //如果多次失败，应该告警
        //     log.debug("解析设备返回的消息失败:{}", e);
        // }
        log.debug("收到设备返回的开始充电结果:{}", chargingResultDto);
        if (chargingResultDto != null) {
            //获取 设备 返回的 开始充电结果
            String result = chargingResultDto.getResult();
            if (ChargingConstants.CHARGING_RESULT_SUCCESS.equals(result)) {
                handleSuccessMessage(chargingResultDto);
            }else {
                handleFailMessage(chargingResultDto);
            }
        }
    }

    /**
     * 处理失败开始充电失败消息
     * 高内聚
     * 低耦合
     * @param chargingResultDto
     */
    private void handleFailMessage(ChargingResultDto chargingResultDto) {
        log.debug("设备发送的消息是 开始充电失败 的消息");
        ChargingOrderFailPO chargingOrderFailPO = new ChargingOrderFailPO();
        chargingOrderFailPO.setBillId(chargingResultDto.getOrderNo());
        chargingOrderFailPO.setFailDesc("设备有响应,开始充电失败");
        log.debug("保存失败订单记录:{}", chargingOrderFailPO);
        int row = mqttContext.getOrderFailMapper().insert(chargingOrderFailPO);
        log.debug("保存失败订单记录:{},影响行数:{}",chargingOrderFailPO,row);
        //todo
        log.debug("设备发送的消息是 开始充电失败 <<通知用户>>,提醒用户更换设备");
        log.debug("给用户发送优惠券，表示歉意，给用户心灵上的安慰");
        log.debug("修改枪的状态为 <<故障>>");
        log.debug("<<通知设备管理员>> 进行设备的检修");
    }

    /**
     * 处理失败开始充电成功消息
     * @param chargingResultDto
     */
    private void handleSuccessMessage(ChargingResultDto chargingResultDto) {
        log.debug("设备发送的消息是 开始充电成功 的消息");
        ChargingOrderSuccessPO successPO = new ChargingOrderSuccessPO();
        successPO.setBillId(chargingResultDto.getOrderNo());
        successPO.setGunId(chargingResultDto.getGunId());
        //订单服务收到设备 开始充电成功 消息 保存成功订单记录 订单的状态 充电中
        successPO.setBillStatus(OrderStatusConstants.ORDER_STATUS_CHARGING);
        log.debug("保存成功订单记录:{}", successPO);
        int row = mqttContext.getOrderSuccessMapper().insert(successPO);
        log.debug("保存成功订单记录:{},影响行数:{}", successPO,row);
        //todo 设备会定时的和订单服务同步充电进度数据 直到车辆电池充满
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("deliveryComplete 消息发送完成:{}", token);
    }


}
